package com.dsk.connector.sink.common.listener;

import com.dsk.connector.commons.conf.TableMapping;
import com.dsk.connector.sink.common.consumer.SinkConsumer;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Component
public class DatatransMqUtils implements ApplicationContextAware {

    private static DatatransMqUtils instance;

    private static ApplicationContext applicationContext;

    public DatatransMqUtils(){
        instance=this;
    }

    @Autowired
    private  Environment environment;

    @Qualifier("tableMappings")
    Map<String, TableMapping> tableMappings;

    @Autowired
    SinkConsumer consumer;

    private String queueNamePrefix="queue_";

    /**
     * 获取配置的线程数
     * @return
     */
    public int getConfigThreadCount(){
        Integer threadCount=environment.getProperty("config.sink.threadCount",Integer.class);
        if(threadCount==null){
            threadCount=this.tableMappings.size();
        }
        if(threadCount> this.tableMappings.size()){
            threadCount= this.tableMappings.size();
        }

        return threadCount;
    }

    /**
     * 创建按表格名称的Listner
     * @return
     */
    private List<MessageListener> createTableMqListener(){

        Integer threadCount=this.getConfigThreadCount();

        List<MessageListener> listeners=new ArrayList<>();

        for(int i=0;i<threadCount;i++){
            MessageListener listener=new MessageListener();
            listener.setConsumer(consumer);
            listeners.add(listener);
        }
        int index=0;
        for(String tableName:tableMappings.keySet()){
            MessageListener listener=listeners.get(index);
            listener.addTopic(tableName);
            index++;
            index=index%threadCount;
        }

        //添加数据库的ddl修改topic
        MessageListener alterLister=new MessageListener();
        alterLister.addTopic("ddl");
        alterLister.addTopic("default");
        alterLister.setConsumer(consumer);

        listeners.add(alterLister);

        return listeners;
    }

    private List<MessageListener> createThreadNumMqListener(){
        int threadCount=this.getConfigThreadCount();

        List<MessageListener> listeners=new ArrayList<>();

        for(int i=0;i<threadCount;i++){
            MessageListener listener=new MessageListener();
            listener.setConsumer(consumer);
            listener.addTopic(queueNamePrefix+i);
            listeners.add(listener);
        }
        return listeners;
    }

    public List<MessageListener> createMqListener(){
        String createQueueBy=this.environment.getProperty("config.source.queueNameBy","threadNum");
        if("threadNum".equalsIgnoreCase(createQueueBy)){
            return this.createThreadNumMqListener();
        }else{
            return this.createTableMqListener();
        }
    }

    /**
     * 获取消息应该发送的queue_name
     * @param sourceTableName
     * @param operation
     * @return
     */
    public String getQueueName(String sourceTableName,String operation){
        String createQueueBy=this.environment.getProperty("config.source.queueNameBy","threadNum");
        String queueName=sourceTableName;
        if("ddl".equalsIgnoreCase(operation)){
            //所有的ddl操作位于同一个队列，保证操作的顺序
            queueName="ddl";
        }
        if(StringUtils.isBlank(queueName)){
            queueName="default";
        }
        if("threadNum".equalsIgnoreCase(createQueueBy)){
            int threadCount=this.getConfigThreadCount();
            int hashCode=Math.abs(queueName.hashCode()%threadCount);
            return queueNamePrefix+hashCode;
        }else{
            return queueName;
        }

    }

    public static DatatransMqUtils getInstance(){
        return instance;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        DatatransMqUtils.applicationContext=applicationContext;
    }

    public Map<String, TableMapping> getTableMappings() {
        return tableMappings;
    }

    public void setTableMappings(Map<String, TableMapping> tableMappings) {
        this.tableMappings = tableMappings;
    }

    public SinkConsumer getConsumer() {
        return consumer;
    }

    public void setConsumer(SinkConsumer consumer) {
        this.consumer = consumer;
    }
}
